package org.apache.ocean.main.replication;

import java.io.File;
import java.io.IOException;
import java.security.MessageDigest;
import java.util.Collections;
import java.util.EnumSet;
import java.util.LinkedList;
import java.util.List;
import java.util.SortedMap;
import java.util.TreeSet;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import java.util.logging.Level;
import java.util.logging.Logger;

import org.apache.ocean.main.replication.LogFile2.Record;
import org.apache.ocean.main.replication.RawLogFile.Data;
import org.apache.ocean.main.replication.RawLogFile.Load;
import org.apache.ocean.main.replication.RawLogFile.RecordHeader;
import org.apache.ocean.main.util.OceanSkipList;
import org.apache.ocean.util.ConcurrentLinkedHashMap;

import com.ibm.icu.text.DecimalFormat;

public class LogFileManager {
	public static Logger log = Logger.getLogger(LogFileManager.class.getName());
	public static final long MAX_FILE_SIZE = 1024 * 1024 * 64;
	private File directoryFile;
	private ConcurrentLinkedHashMap<Long, Record> recordLruMap;
	private OceanSkipList<Long, LogFile2> logFiles = new OceanSkipList<Long, LogFile2>();
	private AtomicLong sequence = new AtomicLong(0);
	private ReentrantReadWriteLock lock = new ReentrantReadWriteLock();
	private ScheduledExecutorService logFileCheckTimer;

	public LogFileManager(File directoryFile) throws IOException {
		this.directoryFile = directoryFile;
		recordLruMap = new ConcurrentLinkedHashMap<Long, Record>(200);
		File[] files = directoryFile.listFiles();
		for (File file : files) {
			if (!file.isDirectory()) {
				long id = getLogFileNumber(file.getName());
				LogFile2 logFile = new LogFile2(id, file);
				logFiles.put(id, logFile);
			}
		}
		if (logFiles.size() > 0) {
			long last = logFiles.lastKey();
			long next = last + 1;
			sequence.set(next);
		}
		logFileCheckTimer = Executors.newSingleThreadScheduledExecutor();
		logFileCheckTimer.scheduleWithFixedDelay(new LogFileSizeCheck(), 1000, 10 * 1000, TimeUnit.MILLISECONDS);
	}
  
	public byte[] getCheckSum(String category, Long lowId, Long highId) throws Exception {
		MessageDigest messageDigest = MessageDigest.getInstance("MD5");
		for (LogFile2 logFile : logFiles.values()) {
			if (!logFile.checksum(category, lowId, highId, messageDigest)) {
				break;
			}
		}
		return messageDigest.digest();
	}
	
	public byte[] getCheckSum(Long lowId, Long highId) throws Exception {
		MessageDigest messageDigest = MessageDigest.getInstance("MD5");
		for (LogFile2 logFile : logFiles.values()) {
			if (!logFile.checksum(lowId, highId, messageDigest)) {
				break;
			}
		}
		return messageDigest.digest();
	}
	
	public class LogFileSizeCheck implements Runnable {
		public void run() {
			try {
				LogFile2 logFile = logFiles.lastValue();
				if (logFile.size() >= MAX_FILE_SIZE) {
					LogFile2 newLogFile = createNewLogFile();
					logFiles.put(newLogFile.getId(), newLogFile);
				}
			} catch (IOException ioException) {
        log.log(Level.SEVERE, "", ioException);
			}
		}
	}

	public boolean delete(Long id) throws Exception {
		LogFile2 logFile = getLogFileContaining(id);
		if (logFile == null)
			throw new Exception("unknown id: " + id);
		return logFile.delete(id);
	}

	public Long getPreviousId(Long id) {
		LogFile2 logFile = getLogFileContaining(id);
		Long previousId = logFile.getPreviousId(id);
		if (previousId == null) {
			SortedMap<Long, LogFile2> subMap = logFiles.headMap(logFile.getId());
			Long lastKey = subMap.lastKey();
			if (lastKey != null) {
				logFile = subMap.get(lastKey);
				previousId = logFile.getPreviousId(id);
			}
		}
		return previousId;
	}

	public Long getHighestId(String category) {
		List<Long> list = new LinkedList<Long>();
		for (LogFile2 logFile : logFiles.values()) {
			list.add(logFile.getHighestID(category));
		}
		return Collections.max(list);
	}

	public Long getLowestId() {
		TreeSet<Long> sortedSet = new TreeSet<Long>();
		for (LogFile2 logFile : logFiles.values()) {
			sortedSet.add(logFile.getLowestId());
		}
		return sortedSet.first();
	}

	public Long getHighestId() {
		TreeSet<Long> sortedSet = new TreeSet<Long>();
		for (LogFile2 logFile : logFiles.values()) {
			sortedSet.add(logFile.getHighestId());
		}
		return sortedSet.last();
	}

	public LogFile2 createNewLogFile() throws IOException {
		long id = sequence.getAndIncrement();
		String fileName = createLogFileName(id);
		File file = new File(directoryFile, fileName);
		LogFile2 logFile = new LogFile2(id, file);
		lock.writeLock().lock();
		try {
			logFiles.put(id, logFile);
			return logFile;
		} finally {
			lock.writeLock().unlock();
		}
	}

	private static String createLogFileName(long value) {
		DecimalFormat decimalFormat = (DecimalFormat) DecimalFormat.getInstance();
		decimalFormat.applyPattern("00000000");
		return "log" + decimalFormat.format(value) + ".bin";
	}

	private static long getLogFileNumber(String fileName) {
		String numberString = fileName.substring(3, fileName.lastIndexOf('.'));
		return Long.parseLong(numberString);
	}

	private LogFile2 getLogFileContaining(Long id) {
		lock.readLock().lock();
		try {
			for (LogFile2 logFile : logFiles.values()) {
				if (logFile.containsId(id)) {
					return logFile;
				}
			}
			return null;
		} finally {
			lock.readLock().unlock();
		}
	}

	public boolean contains(Long id) {
		lock.readLock().lock();
		try {
			for (LogFile2 logFile : logFiles.values()) {
				if (logFile.containsId(id)) {
					return true;
				}
			}
			return false;
		} finally {
			lock.readLock().unlock();
		}
	}

	// TODO: doesn't return greater than
	private LogFile2 getLogFileContaining(String category, Long id) {
		lock.readLock().lock();
		try {
			for (LogFile2 logFile : logFiles.values()) {
				if (logFile.containsId(category, id)) {
					return logFile;
				}
			}
			return null;
		} finally {
			lock.readLock().unlock();
		}
	}

	public List<RecordHeader> getRecordHeaders(Long snapshotID, int num) throws Exception {
		LogFile2 logFile = getLogFileContaining(snapshotID);
		if (logFile == null)
			throw new Exception("unknown snapshotid: " + snapshotID);
		// TODO: handle getting more if needed from next log file
		return logFile.getRecordHeaders(snapshotID, num);
	}

	public Record getRecord(EnumSet<Load> load, RecordHeader recordHeader, ByteBufferPool documentByteBuffers, ByteBufferPool deletesByteBuffers) throws Exception {
		LogFile2 logFile = getLogFileContaining(recordHeader.id);
		if (logFile == null)
			throw new Exception("unknown snapshotid: " + recordHeader.id);
		// TODO: handle getting more if needed from next log file
		return logFile.getRecord(load, recordHeader, documentByteBuffers, deletesByteBuffers);
	}

	public RecordHeader writeRecord(Long id, String category, long schemaVersion, Data data) throws IOException {
		LogFile2 logFile = getCurrentLogFile();
		return logFile.writeRecord(id, category, schemaVersion, data);
	}

	public Record getRecordGreaterOrEqualTo(EnumSet<Load> load, Long snapshotID, ByteBufferPool documentsByteBuffers, ByteBufferPool deletesByteBuffers) throws IOException {
		LogFile2 logFile = getLogFileContaining(snapshotID);
		if (logFile == null)
			return null;
		return logFile.getRecordGreaterOrEqualToID(load, snapshotID, documentsByteBuffers, deletesByteBuffers);
	}

	public Record getRecordGreaterOrEqualTo(EnumSet<Load> load, String category, Long snapshotID, ByteBufferPool documentsByteBuffers, ByteBufferPool deletesByteBuffers) throws IOException {
		LogFile2 logFile = getLogFileContaining(category, snapshotID);
		if (logFile == null)
			return null;
		return logFile.getRecordGreaterOrEqualToID(load, category, snapshotID, documentsByteBuffers, deletesByteBuffers);
	}

	public LogFile2 getCurrentLogFile() throws IOException {
		LogFile2 logFile = logFiles.lastValue();
		if (logFile == null) {
			logFile = createNewLogFile();
		}
		return logFile;
	}
}
